package com.streamsets.datacollector.execution.mq.config;

import com.streamsets.datacollector.bundles.Constants;
import com.streamsets.datacollector.execution.mq.MqStateEventNotifier;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.FileReader;
import java.util.Properties;

/**
* MQ 配置工具类
* @author wengzhonghui
* @date: 2021-07-01 18:04:15
*/
public class MqConfigUtil {

  private static final Logger LOG = LoggerFactory.getLogger(MqConfigUtil.class);

  public static final String BROKER_LIST_NAME = "mq.broker.list";
  public static final String TOPIC_STATE_EVENT_NAME = "mq.topic.stateEvent";
  public static final String TOPIC_METRICS_EVENT_NAME = "mq.topic.metricsEvent";

  public static Properties kafkaProp = null;

  private String baseConfDir;

  public MqConfigUtil(String baseConfDir) {
    this.baseConfDir = baseConfDir;
  }

  private  Properties defaultPipelineProperties = null;

  private  void initDefaultProperties(){

    defaultPipelineProperties = new Properties();
    String filePath = baseConfDir + "/" + Constants.DEFAULT_PIPELINE_CONFIG_NAME;;
    try{
      BufferedReader bufferedReader= new BufferedReader(new FileReader(filePath));
      defaultPipelineProperties.load(bufferedReader);
    }catch (Exception e){
      LOG.error("getTemplateContent error for filePath:" + filePath + "message:" + e.getMessage());
    }
  }
  private  Properties getDefaultPipelineProperties(){
    if(defaultPipelineProperties == null){
      initDefaultProperties();
    }
    return defaultPipelineProperties;
  }

  public String getMqBrokerList(){
    Properties prop = getDefaultPipelineProperties();
    return prop.get(BROKER_LIST_NAME).toString();
  }

  public String getMetricsEventTopicName(){
    Properties prop = getDefaultPipelineProperties();
    return prop.get(TOPIC_METRICS_EVENT_NAME).toString();
  }

  public String getStateEventTopicName(){
    Properties prop = getDefaultPipelineProperties();
    return prop.get(TOPIC_STATE_EVENT_NAME).toString();
  }

  private static KafkaProducer kafkaProducer = null;

  public synchronized KafkaProducer  getKafkaProducer(){
    if(kafkaProducer == null){
      Properties props = initConfig();
      kafkaProducer = new KafkaProducer<>(props);
    }
    return kafkaProducer;
  }

  public  Properties initConfig() {
    if (kafkaProp == null) {
      kafkaProp = new Properties();
      kafkaProp.put("bootstrap.servers", getMqBrokerList());
      kafkaProp.put("key.serializer",
        "org.apache.kafka.common.serialization.StringSerializer");
      kafkaProp.put("value.serializer",
        "org.apache.kafka.common.serialization.StringSerializer");
      // 空字符时，kafka会自动生成一个Id，必须有，不然会报空异常
      kafkaProp.put(ProducerConfig.CLIENT_ID_CONFIG, "streamsets" + System.currentTimeMillis());
    }
    return kafkaProp;
  }


}
